package com.bigdata

import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import com.bigdata.common.constants.Constants
import com.bigdata.common.utils.KafkaCommonUtils
import com.bigdata.common.utils.KafkaCommonUtils.produceMockData
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.util.Random

/**
 * 模拟生成订单事件日志
 */
object KafkaProducerOrder {

  def main(args: Array[String]): Unit = {
    writeToKafka(Constants.Topic_Order)
  }

  def writeToKafka(topic: String): Unit = {
    val producer: KafkaProducer[String, String] = KafkaCommonUtils.createKafkaProducer(topic)
    try {
      while (true) {
        //        [1, "create", 1558430842000L]
        val userId = new Random().nextInt(10000)
        var eventType = "create"
        var createTime = System.currentTimeMillis() - new Random().nextInt(1000000)
        val canPay = new Random().nextBoolean()

        var data = s"$userId,$eventType,$createTime"
        println(s"create: [$data]")
        producer.send(new ProducerRecord[String, String](topic, data))

        if (canPay) {
          eventType = "pay"
          createTime = createTime + new Random().nextInt(5*60*1000)
          data = s"$userId,$eventType,$createTime"
          println(s"pay: [$data]")
          producer.send(new ProducerRecord[String, String](topic, data))
        }

        Thread.sleep(5000)
      }
    } catch {
      case ex: Exception => ex.printStackTrace()
    }
    finally {
      producer.close()
    }

  }


}